package pool

import (
	"errors"
	"fmt"
	"sync"
)

var (
	// ErrNoIdleWorkerInPool workerPool中任务已满，没有空闲goroutine用于处理新任务
	ErrNoIdleWorkerInPool = errors.New("no idle worker in pool")
	// ErrWorkerPoolFreed workerPool已终止运行
	ErrWorkerPoolFreed = errors.New("workerpool freed")
)

const (
	defaultCapacity = 100
	maxCapacity     = 10000
)

// WorkerPool workerPool 数据结构定义
type WorkerPool struct {
	// workerPool大小
	capacity int
	// active channel
	active chan struct{}
	// task channel
	tasks chan Task
	// 用于在pool销毁时等待所有worker退出
	wg sync.WaitGroup
	// 用于通知各个worker退出的信号channel
	quit chan struct{}
}

// New 用于创建一个 pool 类型实例，并将 pool 池的 worker 管理机制运行起来
func New(capacity int) *WorkerPool {
	if capacity <= 0 {
		capacity = defaultCapacity
	}
	if capacity > maxCapacity {
		capacity = maxCapacity
	}

	pool := &WorkerPool{
		capacity: capacity,
		active:   make(chan struct{}, capacity),
		tasks:    make(chan Task),
		quit:     make(chan struct{}),
	}

	fmt.Println("workerPool.start.....")

	go pool.run()

	return pool
}

// 添加一个任务
// 新 worker 的核心，依然是一个基于 for-select 模式的循环语句，在循环体中，新 worker 通过 select 监视 quit 和 tasks 两个 channel。
// 和 run 方法一样，当接收到来自 quit channel 的退出“信号”时，这个 worker 就会结束运行。
// tasks channel 中放置的是用户通过 Schedule 方法提交的请求，新 worker 会从这个 channel 中获取最新的 Task 并运行这个 Task
func (pool *WorkerPool) newWorker(i int) {
	pool.wg.Add(1)

	// 为了防止用户提交的 task 抛出 panic，进而导致整个 workerPool 受到影响，在 worker 代码的开始处，使用了 defer+recover 对 panic 进行捕捉，
	// 捕捉后 worker 也是要退出的，于是还通过<-p.active更新了 worker 计数器。并且一旦 worker goroutine 退出，p.wg.Done 也需要被调用，
	// 这样可以减少 WaitGroup 的 Goroutine 等待数量
	go func() {
		defer func() {
			if err := recover(); err != nil {
				fmt.Printf("worker[%03d]: recover panic[%s] and exit\n", i, err)
				<-pool.active
			}
			pool.wg.Done()
		}()

		fmt.Printf("worker[%03d]: start\n", i)

		for {
			select {
			case <-pool.quit:
				fmt.Printf("worker[%03d]: exit\n", i)
				<-pool.active
				return
			case task := <-pool.tasks:
				fmt.Printf("worker[%03d]: receive a task\n", i)
				task()
			}
		}
	}()
}

// 任务的运行方法
func (pool *WorkerPool) run() {
	idx := 0
	// 循环体中使用 select 监视 Pool 类型实例的两个 channel：quit 和 active
	for {
		select {
		case pool.active <- struct{}{}: // active channel 可写时，run 方法就会创建一个新的 worker Goroutine
			// create a new worker
			idx++
			pool.newWorker(idx)
		case <-pool.quit: // 当接收到来自 quit channel 的退出“信号”时，这个 Goroutine 就会结束运行
			return
		}
	}
}

// Schedule workerPool包的用户通过该方法向 pool 池提交待执行的任务（Task）。
func (pool *WorkerPool) Schedule(task Task) error {

	// Schedule 方法的核心逻辑，是将传入的 Task 实例发送到 workerpool 的 tasks channel 中。
	// 但考虑到现在 workerpool 已经被销毁的状态，这里通过一个 select，检视 quit channel 是否有“信号”可读，如果有，就返回一个哨兵错误 ErrWorkerPoolFreed。
	// 如果没有，一旦 p.tasks 可写，提交的 Task 就会被写入 tasks channel，以供 pool 中的 worker 处理。

	select {
	case <-pool.quit:
		return ErrWorkerPoolFreed
	//  WorkerPool 结构体中的 tasks 是一个无缓冲的 channel，如果 pool 中 worker 数量已达上限，而且 worker 都在处理 task 的状态，
	//  那么 Schedule 方法就会阻塞，直到有 worker 变为 idle 状态来读取 tasks channel，schedule 的调用阻塞才会解除。
	case pool.tasks <- task:
		return nil
	}
}

// Free 用于销毁一个 pool 池，停掉所有 pool 池中的 worker
func (pool *WorkerPool) Free() {
	close(pool.quit)
	pool.wg.Wait()
	fmt.Printf("workerpool freed\n")
}

// Task 任务定义
type Task func()
